package org.example.com.atguigu.day04;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.api.java.UDF3;
import static org.apache.spark.sql.types.DataTypes.StringType;

import org.apache.spark.sql.expressions.UserDefinedFunction;
import org.junit.Test;

import java.io.Serializable;

import static org.apache.spark.sql.functions.udf;

public class SparkSqlTest implements Serializable {
    /*
    *  sparksql的编程风格
    *       1.命名式: 使用一些方法操作数据
    *       2.声明式: 使用sql操作数据[重要]
    * */
    @Test
    public void testCommand(){
        SparkSession spark = SparkSession.builder().master("local[4]").appName("test").getOrCreate();
        Dataset<Row> df = spark.read().option("header", "true").csv("datas/mycsv.csv");
        // df.select("name").show();
        // df.selectExpr("sum(age) sum_age").show();
        df.selectExpr("name","cast (age as bigint) age").groupBy("name").min("age").show();
    }

    @Test
    public void testOption(){
        SparkSession spark = SparkSession.builder().master("local[4]").appName("test").getOrCreate();
        Dataset<Row> df = spark.read().option("header", "true").csv("datas/mycsv.csv");
        // 起表名
        df.createOrReplaceTempView("person");
        spark.sql("select count(*) from person group by name,age").show();
    }

    @Test
    public void testDefinedUDF1(){
        SparkSession spark = SparkSession.builder().master("local[4]").appName("test").getOrCreate();
        Dataset<String> df = spark.read().textFile("datas/user.txt");
        Dataset<Row> ds = df.toDF("id");
        ds.createOrReplaceTempView("user");
        // 定义函数
        UserDefinedFunction uf = udf(new UDF3<String, Integer, String, String>() {
            @Override
            public String call(String id, Integer len, String pad) throws Exception {
                // id = 0010
                // 获取要补的位数
                Integer padlen = len - id.length();
                // 补齐
                for (int i = 0; i < padlen; i++) {
                    id = pad + id;
                }
                return id;
            }
        }, StringType);
        // TODO : 注册到sparksession中
        spark.udf().register("XX", uf);

        spark.sql("select XX(id,8,'-') id from user").show();
    }

}
